package com.atguigu.hbasedmlddl;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;


public class HBase_DDLDML {

    private Configuration conf;
    private Connection connection;
    private Admin admin;

    /**
     * 配置zookeeper通信地址,获取hbase连接和创建master对象
     *
     * @throws IOException
     */
    @Before
    public void connect() throws IOException {

        //1.获取hbase连接对象,连接zk
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop102:2181,hadoop103:2181,hadoop104:2181");
        connection = ConnectionFactory.createConnection(conf);
        //2.获取admin对象
        admin = connection.getAdmin();
    }

    /**
     * 关闭资源
     */
    @After
    public void close() throws IOException {
        //关闭连接
        if (admin != null) {
            admin.close();
        }
        if (connection != null) {
            connection.close();
        }
    }

    /**
     * 创建命名空间
     */
    @Test
    public void createNameSpace() throws IOException {
        //1.封装数据
        NamespaceDescriptor big = NamespaceDescriptor.create("big").build();
        //2.创建命名空间
        admin.createNamespace(big);
    }

    /**
     * 查看所有命名空间
     */
    @Test
    public void list_namespace() throws IOException {
        //1.获取描述器对象
        NamespaceDescriptor[] namespaceDescriptors = admin.listNamespaceDescriptors();
        //2.遍历
        for (NamespaceDescriptor namespaceDescriptor : namespaceDescriptors) {
            System.out.println(namespaceDescriptor.getName());
        }
    }

    /**
     * 查看命名空间所有表
     */
    @Test
    public void list_namespace_table() throws IOException {
        //1.根据命名空间获取表集合
        TableName[] tables = admin.listTableNamesByNamespace("hbase");
        //2.遍历
        for (TableName table : tables) {
            System.out.println(table.getNameAsString());
        }
    }

    /**
     * 删除命名空间
     */
    @Test
    public void deletenamespace() throws IOException {
        //1.根据命名空间获取表集合
        TableName[] tables = admin.listTableNamesByNamespace("big");
        //2.遍历禁用及删除表
        for (TableName table : tables) {
            admin.disableTable(table);
            admin.deleteTable(table);
        }
        //3.删除命名空间
        admin.deleteNamespace("big");
    }
    /**
     * 创建表
     */
    @Test
    public void createTable() throws IOException {
        //1.封装数据:表名列族
        TableName student = TableName.valueOf("test:a2");
        ColumnFamilyDescriptor f1 = ColumnFamilyDescriptorBuilder.newBuilder("f1".getBytes()).build();
        ColumnFamilyDescriptor f2 = ColumnFamilyDescriptorBuilder.newBuilder("f2".getBytes()).build();
        TableDescriptor build = TableDescriptorBuilder.newBuilder(student)
                .setColumnFamily(f1)
                .setColumnFamily(f2)
                .build();
        //2.创建表
        admin.createTable(build);
    }
    @Test
    public void alterTable() throws IOException {
        //1.封装数据:表名列族
        TableName student = TableName.valueOf("student");
        ColumnFamilyDescriptor f1 = ColumnFamilyDescriptorBuilder.newBuilder("f1".getBytes()).setMaxVersions(3).build();
        ColumnFamilyDescriptor f2 = ColumnFamilyDescriptorBuilder.newBuilder("f2".getBytes()).build();
        TableDescriptor build = TableDescriptorBuilder.newBuilder(student)
                .setColumnFamily(f1)
                .setColumnFamily(f2)
                .build();
        //2.修改表
        admin.modifyTable(build);
    }

    /**
     * 插入数据
     */
    @Test
    public void put() throws IOException {
        //1.获取表对象
        TableName student = TableName.valueOf("student");
        Table table = connection.getTable(student);
        //2.封装数据
        Put put = new Put("1001".getBytes());
        put.addColumn("f1".getBytes(),"name".getBytes(),"wzx".getBytes());
        put.addColumn("f1".getBytes(),"age".getBytes(), Bytes.toBytes(20));
        //3.插入
        table.put(put);
        //4.关闭
        table.close();
    }
    @Test
    public void putList() throws IOException {
        //1.获取table对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //2.封装数据
        List<Put> puts = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Put put = new Put(("100"+i).getBytes());
            put.addColumn(("f1").getBytes(),"name".getBytes(),("wzx"+i).getBytes());
            put.addColumn(("f1").getBytes(),"age".getBytes(),Bytes.toBytes(18+i));
            puts.add(put);
        }
        //3.插入
        table.put(puts);
        //4.关闭
        table.close();
    }

    /**
     * 单个主键查询
     */
    @Test
    public void get() throws IOException {
        //1.获取table对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //2.封装数据
        Get get = new Get("1002".getBytes());
        //3.查询
        Result result = table.get(get);
        //4.结果展示
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            String  qualifier= new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            if(family.equals("f1")&&qualifier.equals("age")){
                int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            System.out.println("family="+family+"qualifier="+qualifier+"value="+value);
            }else{
                String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            System.out.println("family="+family+"qualifier="+qualifier+"value="+value);
            }
        }
        //5.关闭
        table.close();
    }
    @Test
    public void getList() throws IOException {
        //1.获取表对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //2.封装数据
        List<Get> gets = new ArrayList<>();
        for (int i = 1; i < 9; i++) {
            Get get = new Get(("100"+i).getBytes());
            //指定查询某一个列族的数据
            //get.addFamily("f1".getBytes());
            //指定查询某一个列限定符的数据
            //get.addColumn("f1".getBytes(),"name".getBytes());
            //指定查询多个版本的数据
            //get.readVersions(3);
            //查询指定时间戳的数据
            //get.setTimestamp(1617792966510L);
            get.setCacheBlocks(true);
            gets.add(get);
        }
        //查找
        Result[] results = table.get(gets);
        //展示结果
        //关闭
        table.close();
    }
    /**
     * 删除表数据
     */
    @Test
    public void delete() throws IOException {
        //1.获取表对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //2.封装操作数据
        Delete delete = new Delete("1002".getBytes());
        delete.addColumn("f1".getBytes(),"age".getBytes());
        //3.进行删除
        table.delete(delete);
        //4.关闭
        table.close();
    }
    /**
     * 清空表数据
     */
    @Test
    public void truncate() throws IOException {
        if(admin.tableExists(TableName.valueOf("student"))){
          admin.disableTable(TableName.valueOf("student"));
            admin.truncateTable(TableName.valueOf("student"),false);
        }

    }
    /**
     * 扫描表
     */
    @Test
    public void scan() throws IOException {
        //1.获取表对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //3.封装数据
        Scan scan = new Scan();
        //查询某个列族数据
//        scan.addFamily("f1".getBytes());
        //查询某个列数据
//        scan.addColumn("f1".getBytes(),"name".getBytes());
        //查询多个版本的数据
//        scan.readVersions(3);
        //查询rowkey范围段的数据
        scan.withStartRow("1002".getBytes(),true);
        scan.withStopRow("1005".getBytes(),true);
        //2.获取扫描对象
        ResultScanner scanner = table.getScanner(scan);
        //4.结果展示
        Iterator<Result> iterator = scanner.iterator();
        while(iterator.hasNext()){
            Result result = iterator.next();
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取rowkey
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                //获取列族
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                //获取列限定符
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                //获取值
                if(family.equals("f1")&&qualifier.equals("age")){
                int value = Bytes.toInt(CellUtil.cloneValue(cell));
                System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }else{
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }
            }
        }
        //5.关闭
        table.close();
    }
    @Test
    public void filter () throws IOException {
        //1.获取表对象
        Table table = connection.getTable(TableName.valueOf("student"));
            //2.1配置扫描对象
        Scan scan = new Scan();
            //2.2配置过滤对象
                //要过滤"name=wzx1"的cell
        SingleColumnValueFilter filter = new SingleColumnValueFilter("f1".getBytes(),"name".getBytes(),CompareOperator.EQUAL,"wzx1".getBytes());
                //设置如果某行数据没有要过滤的列,则不展示该行数据
        filter.setFilterIfMissing(true);
        scan.setFilter(filter);
        //2.获取扫描对象
        ResultScanner scanner = table.getScanner(scan);
        //3.结果展示
        Iterator<Result> iterator = scanner.iterator();
        while(iterator.hasNext()){
            Result result = iterator.next();
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取rowkey
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                //获取列族
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                //获取列限定符
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                //获取值
                if(family.equals("f1")&&qualifier.equals("age")){
                    int value = Bytes.toInt(CellUtil.cloneValue(cell));
                    System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }else{
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }
            }
        }
        //关闭
        table.close();
    }
    @Test
    public void filterlike() throws IOException {
        //1.获取表对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //2.1配置扫描对象
        Scan scan = new Scan();
        //2.2配置过滤器(like模糊过滤) 过滤 f1:name = 'zhangsan-7' 的数据
        SingleColumnValueFilter filter = new SingleColumnValueFilter("f1".getBytes(),"age".getBytes(),CompareOperator.EQUAL,
                new SubstringComparator("zhangsan"));
        scan.setFilter(filter);
        //2.获取扫描对象
        ResultScanner scanner = table.getScanner(scan);
        //3.结果展示
        Iterator<Result> iterator = scanner.iterator();
        while(iterator.hasNext()){
            Result result = iterator.next();
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取rowkey
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                //获取列族
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                //获取列限定符
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                //获取值
                if(family.equals("f1")&&qualifier.equals("age")){
                    int value = Bytes.toInt(CellUtil.cloneValue(cell));
                    System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }else{
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }
            }
        }
        //关闭
        table.close();
    }
    @Test
    public void filterand() throws IOException {

        //1.获取表对象
        Table table = connection.getTable(TableName.valueOf("student"));
        //2.进行过滤查找 过滤(wzx5 and 23) or age = 21
        Scan scan = new Scan();
            //wzx5
        SingleColumnValueFilter filter = new SingleColumnValueFilter("f1".getBytes(),
                "name".getBytes(),CompareOperator.EQUAL,
                "wzx5".getBytes());
            //age=23
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter("f1".getBytes(),
                "age".getBytes(),CompareOperator.EQUAL,
                Bytes.toBytes(23));
            //age=23
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter("f1".getBytes(),
                "age".getBytes(),CompareOperator.EQUAL,
                Bytes.toBytes(21));
        //2.1组合过滤器,配置过滤策略
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        FilterList filterList1 = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        //2.2进行组合
        filterList1.addFilter(filter2);
        filterList.addFilter(filter);
        filterList.addFilter(filter1);
        filterList1.addFilter(filterList);
        scan.setFilter(filterList1);
        ResultScanner scanner = table.getScanner(scan);
        //结果展示
        Iterator<Result> iterator = scanner.iterator();
        while(iterator.hasNext()){
            Result result = iterator.next();
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取rowkey
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                //获取列族
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                //获取列限定符
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                //获取值
                if(family.equals("f1")&&qualifier.equals("age")){
                    int value = Bytes.toInt(CellUtil.cloneValue(cell));
                    System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }else{
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("rowkey:"+rowKey+",family:"+family+",qualifier:"+qualifier+",value:"+value);
                }
            }
        }
        //关闭
        table.close();
    }
}
